package container.queue;

import lombok.AllArgsConstructor;
import util.ThreadUtil;

import java.util.Random;
import java.util.concurrent.SynchronousQueue;

/**
 * 一种无缓冲的等待队列，类似于无中介的直接交易，有点像原始社会中的生产者和消费者，生产者拿着产品去集市销售给产品的最终消费者，而消费者必须亲自去集市
 * 找到所要商品的直接生产者，如果一方没有找到合适的目标，那么对不起，大家都在集市等待。
 * <p>
 * 相对于有缓冲的 BlockingQueue 来说，少了一个中间经销商的环节（缓冲区），如果有经销商，生产者直接把产品批发给经销商，而无需在意经销商最终会将这
 * 些产品卖给那些消费者，由于经销商可以库存一部分商品，因此相对于直接交易模式，总体来说采用中间经销商的模式会吞吐量高一些（可以批量买卖），
 * 但另一方面，又因为经销商的引入，使得产品从生产者到消费者中间增加了额外的交易环节，单个产品的及时响应性能可能会降低。
 * <p>
 * 声明一个 SynchronousQueue 有两种不同的方式，它们之间有着不太一样的行为。公平模式和非公平模式的区别:
 * <pre>
 * public SynchronousQueue() {
 *     this(false);
 * }
 * public SynchronousQueue(boolean fair) {
 *     transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
 * }
 * </pre>
 * 如果采用公平模式：SynchronousQueue 会采用公平锁，并配合一个 FIFO 队列来阻塞多余的生产者和消费者，从而体系整体的公平策略；
 * 如果是非公平模式（默认）：SynchronousQueue 采用非公平锁，同时配合一个 LIFO 队列来管理多余的生产者和消费者，而后一种模式，如果生产者和消费者的处理速度有差距，则很容易出现饥渴的情况，即可能有某些生产者或者是消费者的数据永远都得不到处理。
 * <p>
 * 你不能在 SynchronousQueue 中使用 peek 方法（在这里这个方法直接返回 null），peek 方法的语义是只读取不移除，显然，这个方法的语义是
 * 不符合 SynchronousQueue 的特征的。SynchronousQueue 也不能被迭代，因为根本就没有元素可以拿来迭代的。虽然 SynchronousQueue 间接地
 * 实现了 Collection 接口，但是如果将其当做 Collection 来用的话，那么集合是空的。
 * <p>
 * SynchronousQueue 可以看成是一个传球手，负责把生产者线程处理的数据直接传递给消费者线程。队列本身并不存储任何元素，非常适合传递性场景。
 * SynchronousQueue 的吞吐量高于 ArrayBlockingQueue 和 LinkedBlockingQueue。
 */
public class T07_SynchronusQueue {
    /**
     * 使用 SynchronousQueue 演示生产者、消费者模型
     * 从结果中可以看出如果已经生产但是还未消费的，那么会阻塞在生产一直等到消费才能生成下一个。
     * 这是单线程版本，若是多线程版本，结果相信大家也能猜到。请大家自己尝试书写
     */
    public static void main(String[] args) {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>();
        new Customer(queue).start();
        new Product(queue).start();
    }

    @AllArgsConstructor
    static class Product extends Thread {
        SynchronousQueue<Integer> queue;

        @Override
        public void run() {
            while (true) {
                int rand = new Random().nextInt(1000);
                System.out.println("生产了一个产品：" + rand);
                System.out.println("等待三秒后运送出去...");
                ThreadUtil.sleepSeconds(3);
                queue.offer(rand);
                System.out.println("产品生成完成：" + rand);
            }
        }
    }

    @AllArgsConstructor
    static class Customer extends Thread {
        SynchronousQueue<Integer> queue;

        @Override
        public void run() {
            while (true) {
                try {
                    System.out.println("消费了一个产品:" + queue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("------------------------------------------");
            }
        }
    }
}
